package simpledb.index.hash;

import simpledb.index.Index;
import simpledb.query.Constant;
import simpledb.query.Plan;
import simpledb.query.StringConstant;
import simpledb.query.TablePlan;
import simpledb.query.TableScan;
import simpledb.record.*;
import simpledb.tx.Transaction;

import java.io.File;
import java.util.ArrayList;


/**
 * A static hash implementation of the Index interface.
 * A fixed number of buckets is allocated (currently, 100),
 * and each bucket is implemented as a file of index records.
 *
 * @author Edward Sciore
 */
public class LinearIndex implements Index {
    public int numBuckets = 1;
    private String idxname;
    private Schema sch;
    private Transaction tx;
    private Constant searchkey = null;
    private TableScan ts = null;
    LinearHashParams lhp;
    TableInfo ti;
    private int INIT_NUM_BUCKETS= 1;
    private String indexFileName;

    private static int expCount = 0 ;

    /**
     * Opens a hash index for the specified index.
     *
     * @param idxname the name of the index
     * @param sch     the schema of the index records
     * @param tx      the calling transaction
     */
    public LinearIndex(String idxname, Schema sch, Transaction tx) {
        this.idxname = idxname;
        this.sch = sch;
        this.tx = tx;
        this.lhp = new LinearHashParams();

    }
    
    public LinearIndex(String idxname, Schema sch, Transaction tx, String indexFileName) {
		this(idxname,sch,tx);
		this.indexFileName = indexFileName;
	}
    
    public LinearIndex(String idxname, Schema sch, Transaction tx, String indexFileName, int exConstant) {
		this(idxname,sch,tx,indexFileName);
		this.lhp = new LinearHashParams(0,0,exConstant);
	}

    /**
     * Positions the index before the first index record
     * having the specified search key.
     * The method hashes the search key to determine the bucket,
     * and then opens a table scan on the file
     * corresponding to the bucket.
     * The table scan for the previous bucket (if any) is closed.
     *
     * @see simpledb.index.Index#beforeFirst(simpledb.query.Constant)
     */
    public void beforeFirst(Constant searchkey) {
        close();
        this.searchkey = searchkey;
        int bucket = getBucketValue(searchkey);
        String tblname = idxname + bucket;
        ti = new TableInfo(tblname, sch);
        ts = new TableScan(ti, tx);
    }

    /**
     * Moves to the next record having the search key.
     * The method loops through the table scan for the bucket,
     * looking for a matching record, and returning false
     * if there are no more such records.
     *
     * @see simpledb.index.Index#next()
     */
    public boolean next() {
        while (ts.next())
            if (ts.getVal("dataval").equals(searchkey))
                return true;
        return false;
    }

    /**
     * Retrieves the dataRID from the current record
     * in the table scan for the bucket.
     *
     * @see simpledb.index.Index#getDataRid()
     */
    public RID getDataRid() {
        int blknum = ts.getInt("block");
        int id = ts.getInt("id");
        return new RID(blknum, id);
    }

    /**
     * Inserts a new record into the table scan for the bucket.
     *
     * @see simpledb.index.Index#insert(simpledb.query.Constant, simpledb.record.RID)
     */
    public void insert(Constant val, RID rid) {

        beforeFirst(val);

        int bucketValue = getBucketValue(val);
        /*Uncomment the one below if needed*/
        //System.out.println("\t{ " + val + " , " + val.hashCode() + " }  --- Bucket [" + bucketValue + "]");
        ts.insert();
        ts.setInt("block", rid.blockNumber());
        ts.setInt("id", rid.id());
        ts.setVal("dataval", val);
        
        int lastBlockNumber=getLastBlockNumber(ti, tx);
        ts.close();
        boolean callSplit=false;
		if (lastBlockNumber>0) {
			int recordCount = getRecordCount(ti, tx,lastBlockNumber);
			if (recordCount >= lhp.getExpandConstant()) {
				 callSplit=true;
            }
        } 
		if(callSplit){
			//System.out.println("\n\t************** NEED TO SPLIT **************");
			splitNextBucket();
		}
      
        
    }

    
    private void splitNextBucket() {
    	String decorate = "#############";
    	expCount ++;
    	System.out.println("\n\n\t" + decorate + " BEFORE EXPANSION " + expCount + " " + decorate + "\n");
    	printAllBuckets();
    	System.out.println("\n\t" + decorate + " BEFORE EXPANSION " + expCount + " " + decorate +"\n\n");
        ArrayList<LinearHashEntry> linearHashEntries = entriesToBeRearranged(lhp.getNext());
        
        lhp.setLevel(lhp.getLevel()+1);
        for (LinearHashEntry linearHashEntry : linearHashEntries) {        	
            insert(new StringConstant(linearHashEntry.getDataval()), new RID(linearHashEntry.getBlock(), linearHashEntry.getId()));
        }
        lhp.setLevel(lhp.getLevel()-1);
        lhp.setNext((lhp.getNext() + 1)%(int)(INIT_NUM_BUCKETS * Math.pow(2, lhp.getLevel() )));
   
        
        if(lhp.getNext()==0){
        	lhp.setLevel(lhp.getLevel()+1);
        }
        
        System.out.println("\n\n\t" + decorate + " AFTER EXPANSION " + expCount + " " + decorate + "\n");
        printAllBuckets();
        System.out.println("\n\t" + decorate + " AFTER EXPANSION " + expCount + " " + decorate +"\n\n");
    }

    private ArrayList<LinearHashEntry> entriesToBeRearranged(int next) {
        ArrayList<LinearHashEntry> linearHashEntries = new ArrayList<LinearHashEntry>();
        int nextBucket = next;
        TableInfo splitTi = new TableInfo(idxname + nextBucket, sch);

        TableScan splitTs = new TableScan(splitTi, tx);
        splitTs.beforeFirst();
        while (splitTs.next()) {
            String dataval = splitTs.getString("dataval");
            int block = splitTs.getInt("block");
            int id = splitTs.getInt("id");
            linearHashEntries.add(new LinearHashEntry(block, id, dataval));
            splitTs.delete();
        }
        splitTs.close();
        return linearHashEntries;
    }

    /**
     * Deletes the specified record from the table scan for
     * the bucket.  The method starts at the beginning of the
     * scan, and loops through the records until the
     * specified record is found.
     *
     * @see simpledb.index.Index#delete(simpledb.query.Constant, simpledb.record.RID)
     */
    public void delete(Constant val, RID rid) {
        beforeFirst(val);
        while (next())
            if (getDataRid().equals(rid)) {
                ts.delete();
                return;
            }
    }

    /**
     * Closes the index by closing the current table scan.
     *
     * @see simpledb.index.Index#close()
     */
    public void close() {
        if (ts != null)
            ts.close();
    }

    /**
     * Returns the cost of searching an index file having the
     * specified number of blocks.
     * The method assumes that all buckets are about the
     * same size, and so the cost is simply the size of
     * the bucket.
     *
     * @param numblocks the number of blocks of index records
     * @param rpb       the number of records per block (not used here)
     * @return the cost of traversing the index
     */
    public static int searchCost(int numblocks, int rpb) {
        return numblocks / HashIndex.NUM_BUCKETS;
    }

    public int getNumBuckets() {
        return (int) (numBuckets * Math.pow(2, lhp.getLevel() + 1));
    }

    private int getBucketValue(Constant searchConstant) {
        int level = this.lhp.getLevel();
        int next = this.lhp.getNext();
        int bucket = (int) (Math.abs(searchConstant.hashCode()) % (INIT_NUM_BUCKETS * Math.pow(2, level)));

        if (bucket < next) {
            bucket = (int) (Math.abs(searchConstant.hashCode()) % (INIT_NUM_BUCKETS * Math.pow(2, level + 1)));
        }
        return bucket;
    }

    private int getRecordCount(TableInfo ti, Transaction tx,int blockNum) {
        RecordFile rf = new RecordFile(ti, tx);
        rf.beforeFirst();
        int recordCount = 0;
        while (rf.next()) {
        	if(rf.getCurrentblknum()==blockNum)
            recordCount++;
        }
        rf.beforeFirst();
        rf.close();
        return recordCount;
    }

    private int getLastBlockNumber(TableInfo ti, Transaction tx) {
        RecordFile rf = new RecordFile(ti, tx);
        rf.beforeFirst();
        while (rf.next()) {}
       
        int blockNumber=rf.getCurrentblknum();
        rf.beforeFirst();
        rf.close();
        return blockNumber;
    }
    
    public void printAllBuckets() {
    	String decorate = "----------------------------------------------------------------";   	
    	System.out.println("\t" + decorate);
    	System.out.println("\t" + 
    				"Bucket [#] :          RECORD LIST             -->   OVERFLOW");
    	System.out.println("\t" + decorate);
        boolean proceed = true;
        int i = 0;
        while (proceed) {
            File file = new
            		File(this.indexFileName + i + ".tbl");
            proceed = file.exists();
            if (proceed) {
                printBucket("linearHashIdxTest" + i);
            }
            i++;
        }
        
        System.out.println("\t" + decorate);
    	System.out.println("\t" + "LEVEL: "+ lhp.getLevel() +"\t NEXT: " + lhp.getNext() + "\t EXPAND CONSTANT: " 
    			+ lhp.getExpandConstant());
    	System.out.println("\t" + decorate);
    }

	private void printBucket(String tableName) {
		Plan p = new TablePlan("messy", tx);
	    TableScan s1 = (TableScan) p.open();
		System.out.print("\t" + "Bucket [" + tableName.substring(this.idxname.length()) + "] : " );
		ArrayList<String> entries = new ArrayList<String>();
	    RecordFile recordFile = new RecordFile(new
	    		   TableInfo(tableName, sch), tx);
	    while (recordFile.next()) {
	    	int id = recordFile.getInt("id");
	        int block = recordFile.getInt("block");
	        RID rid = new RID(block, id);
	        s1.beforeFirst();
	        while (s1.next())
	        	if (s1.getRid().equals(rid)) {
	        		String col2 = s1.getString("col2");
	        		entries.add(col2.substring(col2.length() - 3));
	        	}	                     
	       }
	       
	    	int itr;
	    	for(itr = 0 ; itr < entries.size() && itr <5; itr ++ ) {
	    		System.out.print("| " + entries.get(itr) + " ");
	    	}
	    	
	    	
	    	if(itr < entries.size()) {
	    		System.out.print("|  -->  ");
	    		for (;itr<entries.size(); itr++) {
	    			System.out.print("| " + entries.get(itr) + " ");
	    		}
	    	}
	    	System.out.println("|");
	       recordFile.close();
	       s1.close();	
	}
	
	public void setINIT_NUM_BUCKETS(int iNIT_NUM_BUCKETS) {
		INIT_NUM_BUCKETS = iNIT_NUM_BUCKETS;
	}
	
}
